

package com.nageoffer.onecoupon.framework.idempotent;

import com.nageoffer.onecoupon.framework.exception.ServiceException;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.aspectj.lang.ProceedingJoinPoint;
import org.aspectj.lang.annotation.Around;
import org.aspectj.lang.annotation.Aspect;
import org.aspectj.lang.reflect.MethodSignature;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.data.redis.core.script.RedisScript;

import java.lang.reflect.Method;
import java.util.List;
import java.util.Objects;
import java.util.concurrent.TimeUnit;

/**
 * 防止消息队列消费者重复消费消息切面控制器
 */
@Slf4j
@Aspect
@RequiredArgsConstructor
public final class NoMQDuplicateConsumeAspect {

    private final StringRedisTemplate stringRedisTemplate;

    /**
     * 'NX':当key不存在时设置。
     * 'GET'：返回旧值；无论设置是否成功，都会返回键的旧值
     *        如果键是存在的，则插入失败并返回旧值。
     *        如果键不存在，则插入成功并返回nil
     * 'PX'：设置过期时间
     */
    private static final String LUA_SCRIPT = """
            local key = KEYS[1]
            local value = ARGV[1]
            local expire_time_ms = ARGV[2]
            return redis.call('SET', key, value, 'NX', 'GET', 'PX', expire_time_ms)
            """;

    /**
     * 增强方法标记 {@link NoMQDuplicateConsume} 注解逻辑
     */
    @Around("@annotation(com.nageoffer.onecoupon.framework.idempotent.NoMQDuplicateConsume)")
    public Object noMQRepeatConsume(ProceedingJoinPoint joinPoint) throws Throwable {
        NoMQDuplicateConsume noMQDuplicateConsume = getNoMQDuplicateConsumeAnnotation(joinPoint);
        String uniqueKey = noMQDuplicateConsume.keyPrefix() + SpELUtil.parseKey(noMQDuplicateConsume.key(), ((MethodSignature) joinPoint.getSignature()).getMethod(), joinPoint.getArgs());

        String absentAndGet = stringRedisTemplate.execute(
                RedisScript.of(LUA_SCRIPT, String.class),
                List.of(uniqueKey), //redis的key。唯一的KEY，与具体的业务相关联
                IdempotentMQConsumeStatusEnum.CONSUMING.getCode(), //redis的value。目的是标记消息为消费中：0；如果已消费则为1。
                String.valueOf(TimeUnit.SECONDS.toMillis(noMQDuplicateConsume.keyTimeout())) //将秒传换成毫秒
        );

        /**
         * LUA脚本返回值如果不为空，说明不是首次消费，消息的状态可能是消费中或者是已消费。
         *  分情况讨论：
         * Ⅰ、如果消息的状态为消费中，说明消息在消费的过程中出现异常，直接throw抛异常，让MQ进行重试。
         * Ⅱ、如果消息的状态为已消费，说明消息消费者成功消费，直接return返回即可，不执行原业务逻辑。
         */
        if (Objects.nonNull(absentAndGet)) {
            boolean errorFlag = IdempotentMQConsumeStatusEnum.isError(absentAndGet);
            log.warn("[{}] MQ repeated consumption, {}.", uniqueKey, errorFlag ? "Wait for the client to delay consumption" : "Status is completed");
            if (errorFlag) {
                //消费中
                //在 throw 抛出异常后的 return 语句不会被执行，方法终止!
                throw new ServiceException(String.format("消息消费者幂等异常，幂等标识：%s", uniqueKey));
            }
            //已消费
            return null;
        }

        //LUA脚本执行后返回值为nil说明是首次消费消息，执行原业务逻辑。
        Object result;
        try {
            // 执行标记了消息队列防重复消费注解的方法原逻辑
            result = joinPoint.proceed();

            /**
             * 消息消费的过程中没有出现异常说明消息消费成功，将消息的状态设置为已消费，
             * 并设置防重令牌 Key 过期时间，单位秒。
              */
            stringRedisTemplate.opsForValue().set(uniqueKey, IdempotentMQConsumeStatusEnum.CONSUMED.getCode(), noMQDuplicateConsume.keyTimeout(), TimeUnit.SECONDS);
        } catch (Throwable ex) {
            // 删除幂等 Key，让消息队列消费者重试逻辑进行重新消费
            stringRedisTemplate.delete(uniqueKey);
            throw ex;
        }
        return result;
    }

    /**
     * @return 返回自定义防重复消费注解
     */
    public static NoMQDuplicateConsume getNoMQDuplicateConsumeAnnotation(ProceedingJoinPoint joinPoint) throws NoSuchMethodException {
        //1.获得方法签名
        MethodSignature methodSignature = (MethodSignature) joinPoint.getSignature();

        //2.通过反射获取Method类
        Method targetMethod = joinPoint.getTarget().getClass().getDeclaredMethod(methodSignature.getName(), methodSignature.getMethod().getParameterTypes());

        //3.返回方法的注解
        return targetMethod.getAnnotation(NoMQDuplicateConsume.class);
    }
}
